/**
 * 基于间隔的join
 */
package com.atguigu.day5

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object IntervalJoinExceple {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //    设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val clickStrem = env.fromElements(
      ("1", "click", 3600 * 1000L)
    ).assignAscendingTimestamps(_._3)
      .keyBy(_._1)

    val browseStream = env.fromElements(
      ("1", "browse", 2000 * 1000L),
      ("1", "browse", 3100 * 1000L),
      ("1", "browse", 3200 * 1000L)
    ).assignAscendingTimestamps(_._3)
        .keyBy(_._1)
    clickStrem
      .intervalJoin(browseStream)
      //join区间
      .between(Time.minutes(-10),Time.minutes(0))
      .process(new MyInterJoin)
      .print()




    browseStream
      .intervalJoin(clickStrem)
      .between(Time.minutes(0),Time.minutes(10))
      .process(new MyInterJoin)
      .print()

    env.execute()





  }

  //分流开窗后，属于同一个窗口的input1中的元素的input2中的元素做笛卡儿积
  //相同的key，而且是相同的窗口，中的元素做笛卡儿积
  class MyInterJoin extends ProcessJoinFunction[(String,String,Long),(String,String,Long),String]{
    override def processElement(left: (String, String, Long), right: (String, String, Long), ctx: ProcessJoinFunction[(String, String, Long), (String, String, Long), String]#Context, out: Collector[String]): Unit = {
      out.collect(left+"========>"+right)
    }
  }

}
